/*
 *    Copyright 2022 The DSMS Authors.
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */

package com.dsms.dfsbroker.common.api;


import cn.hutool.core.util.ArrayUtil;
import com.alibaba.fastjson2.JSON;
import com.dsms.common.constant.RemoteResponseStatusEnum;
import com.dsms.common.constant.ResultCode;
import com.dsms.common.exception.DsmsEngineException;
import com.dsms.common.remotecall.model.FinishedDetail;
import com.dsms.common.remotecall.model.RemoteRequest;
import com.dsms.common.remotecall.model.RemoteResponse;
import com.dsms.common.remotecall.service.IRemoteCall;
import com.dsms.dfsbroker.common.CommandDirector;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpMethod;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;


@Component
@Slf4j
public class CommonApi {

    /**
     * dsms-storage request retry time interval(milliseconds)
     */
    private static final long INTERVAL_TIME = 200;

    /**
     * dsms-storage request retry times
     */
    public static final int DEFAULT_RETRY_TIMES = 15;
    /**
     * dsms-storage request constant retry threshold.
     * e.g. 200ms, 400ms, 600ms, 600ms, 600m, 600m...
     */
    private static final int CONSTANT_RETRY_THRESHOLD = 3;

    @Autowired
    private IRemoteCall remoteCall;

    public <T> T remoteCallRequest(RemoteRequest remoteRequest, int maxRetries, Class<T> type) throws Throwable {
        remoteRequest.setUrlPrefix("request");
        remoteRequest.setHttpMethod(HttpMethod.POST);
        RemoteResponse remoteResponse = remoteCall(remoteRequest, maxRetries, RemoteResponse.class);
        RemoteResponse response = getRequestResult(remoteRequest, maxRetries, remoteResponse.getId());
        if (Objects.equals(response.getState(), RemoteResponseStatusEnum.SUCCESS.getMessage())) {
            if (!response.getIsFinished()) {
                log.error("get remote call finished result failed. request:{},response:{},retryTime:{}", JSON.toJSONString(remoteRequest), JSON.toJSONString(response), maxRetries);
                throw new DsmsEngineException(ResultCode.REMOTE_CALL_FAIL);
            }
            List<FinishedDetail> finished = response.getFinished();
            String out = StringUtils.hasText(finished.get(0).getOutb()) ? finished.get(0).getOutb() : finished.get(0).getOuts();

            if(String.class.isAssignableFrom(type)){
                return (T) out;
            }else {
                return JSON.parseObject(out, type);
            }
        } else if (Objects.equals(response.getState(), RemoteResponseStatusEnum.FAILED.getMessage())) {
            log.error("get remote call result failed. request:{},response:{},retryTime:{}", JSON.toJSONString(remoteRequest), JSON.toJSONString(response), maxRetries);
            throw new DsmsEngineException(ResultCode.REMOTE_CALL_FAIL);
        } else if (Objects.equals(response.getState(), RemoteResponseStatusEnum.PENDING.getMessage())) {
            log.error("get remote call result timeout. request:{},response:{},retryTime:{}", JSON.toJSONString(remoteRequest), JSON.toJSONString(response), maxRetries);
            throw new DsmsEngineException(ResultCode.REMOTE_CALL_TIMEOUT);
        } else {
            log.error("get remote call result unknown. request:{},response:{},retryTime:{}", JSON.toJSONString(remoteRequest), JSON.toJSONString(response), maxRetries);
            throw new DsmsEngineException(ResultCode.REMOTE_CALL_UNKNOWN);
        }
    }

    public <T> T remoteCall(RemoteRequest remoteRequest, Integer retryTime, Class<T> type) throws Throwable {

        for (int retry = 1; retry <= retryTime; retry++) {
            RemoteResponse remoteResponse = remoteCall.callRestfulApi(remoteRequest);
            T result = JSON.parseObject(remoteResponse.getBody(), type);
            if (ArrayUtil.isArray(result)) {
                return result;
            }
            if (!ObjectUtils.isEmpty(result)) {
                return result;
            }
            TimeUnit.SECONDS.sleep(1);
        }
        return null;
    }


    /**
     * get the result of GET /request
     *
     * @return RemoteResponse
     */
    public RemoteResponse getRequestResult(RemoteRequest remoteRequest, Integer retryTime, String requestId) throws Throwable {
        CommandDirector.constructGetRequest(remoteRequest, requestId);
        RemoteResponse remoteResponse = null;
        for (int retry = 1; retry <= retryTime; retry++) {
            remoteResponse = remoteCall(remoteRequest, retryTime, RemoteResponse.class);
            if (!remoteResponse.getIsFinished()) {
                long timeout = INTERVAL_TIME * retry;
                if (retry > CONSTANT_RETRY_THRESHOLD) {
                    timeout = INTERVAL_TIME * CONSTANT_RETRY_THRESHOLD;
                }
                TimeUnit.MILLISECONDS.sleep(timeout);
            } else {
                break;
            }
        }

        return remoteResponse;
    }

}
